package com.shanks.framework.mq.monitor;

import com.shanks.framework.mq.common.util.JsonUtil;
import com.shanks.framework.mq.common.util.MDCUtil;
import com.shanks.framework.mq.enums.MessageRecordStatusEnum;
import com.shanks.framework.mq.model.MessageRecordPO;
import com.shanks.framework.mq.queue.MessageQueueManager;
import com.shanks.framework.mq.service.MessagePushService;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * 监控处理发送中消息
 */
@Component
public class MessageSendingMonitor {

    private static Logger logger = LoggerFactory.getLogger(MessageSendingMonitor.class);

    @Resource
    private MessageQueueManager messageQueueManager;

    @Resource
    private MessagePushService messagePushService;


    public void monitorSendingMessages() {
        while (true) {
            try {
                MDCUtil.putTraceId();
                MessageRecordPO record = messageQueueManager.takeMessage(MessageRecordStatusEnum.SENDING);
                if (record != null) {
                    long startTime = System.currentTimeMillis();
                    Integer queueSize = messageQueueManager.getQueueSize(MessageRecordStatusEnum.SENDING);
                    logger.debug("===================sending队列,当前队列长度 " + queueSize + ", ID: " + record.getMessageId() + "，消息：" + JsonUtil.allToJson(record));
                    // 发送消息
                    messagePushService.pushMessage(record, MessageRecordStatusEnum.SENDING);
                    queueSize = messageQueueManager.getQueueSize(MessageRecordStatusEnum.SENDING);
                    logger.debug("===================sending队列,当前队列长度 " + queueSize + ", ID: " + record.getMessageId() + "，当前消息耗时：" + (System.currentTimeMillis() - startTime) + "ms");
                }
            } catch (Exception e) {
                logger.error("持续监听sending队列出现异常 : " + e);
            } finally {
                MDCUtil.removeTraceId();
            }
        }
    }
}
